package com.tang.deserialization;

import com.alibaba.fastjson2.JSONObject;
import com.ververica.cdc.debezium.DebeziumDeserializationSchema;
import io.debezium.data.Envelope;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.util.Collector;
import org.apache.kafka.connect.data.Field;
import org.apache.kafka.connect.data.Schema;
import org.apache.kafka.connect.data.Struct;
import org.apache.kafka.connect.data.Timestamp;
import org.apache.kafka.connect.source.SourceRecord;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.format.DateTimeFormatter;
import java.util.List;

/**
 * 自定义序列化器
 * 
 * @author tang
 */
public class CustomDeserialization implements DebeziumDeserializationSchema<String> {

    @Override
    public void deserialize(SourceRecord sourceRecord, Collector<String> collector) {
        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");
        JSONObject res = new JSONObject();

        // 获取数据库和表名称
        String topic = sourceRecord.topic();
        String[] fields = topic.split("\\.");
        JSONObject source = new JSONObject();
        source.put("name", fields[0]);
        source.put("db", fields[1]);
        source.put("table", fields[2]);
        source.put("connector", fields[0].split("_")[0]);

        Struct value = (Struct) sourceRecord.value();
        // 获取before数据
        Struct before = value.getStruct("before");
        JSONObject beforeJson = new JSONObject();
        if (before != null) {
            Schema beforeSchema = before.schema();
            List<Field> beforeFields = beforeSchema.fields();
            for (Field field : beforeFields) {
                Object beforeValue = before.get(field);
                if (Schema.Type.INT64.equals(field.schema().type()) && "io.debezium.time.Timestamp".equals(field.schema().name())) {
                    if (beforeValue != null) {
                        Instant instant = Instant.ofEpochMilli((Long) beforeValue);
                        LocalDateTime dateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
                        String dateTimeStr = formatter.format(dateTime);

                        beforeJson.put(field.name(), dateTimeStr);
                    } else {
                        beforeJson.put(field.name(), beforeValue);
                    }
                } else {
                    beforeJson.put(field.name(), beforeValue);
                }

            }
        }

        // 获取after数据
        Struct after = value.getStruct("after");
        JSONObject afterJson = new JSONObject();
        if (after != null) {
            Schema afterSchema = after.schema();
            List<Field> afterFields = afterSchema.fields();
            for (Field field : afterFields) {
                Object afterValue = after.get(field);
                if (Schema.Type.INT64.equals(field.schema().type()) && "io.debezium.time.Timestamp".equals(field.schema().name())) {
                    if (afterValue != null) {
                        Instant instant = Instant.ofEpochMilli((Long) afterValue);
                        LocalDateTime dateTime = LocalDateTime.ofInstant(instant, ZoneId.systemDefault());
                        String dateTimeStr = formatter.format(dateTime);
                        afterJson.put(field.name(), dateTimeStr);
                    } else {
                        afterJson.put(field.name(), afterValue);
                    }
                } else {
                    afterJson.put(field.name(), afterValue);
                }
            }
        }

        Envelope.Operation operation = Envelope.operationFor(sourceRecord);
        // 将字段写到json对象中
        res.put("before", beforeJson);
        res.put("after", afterJson);
        res.put("source", source);
        res.put("op", operation.code());
        res.put("ts_ms", Timestamp.builder().defaultValue());
        res.put("transaction", null);

        //输出数据
        collector.collect(res.toString());
    }

    @Override
    public TypeInformation<String> getProducedType() {
        return BasicTypeInfo.STRING_TYPE_INFO;
    }
}